package com.flute.haflute.agent.topo;

import gnu.cajo.invoke.Remote;

import java.util.Map;
import java.util.Map.Entry;
import java.util.Timer;
import java.util.TimerTask;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.agent.AgentContext;
import com.flute.haflute.common.ClusterConstants;
import com.flute.haflute.common.ClusterConstants.ROLE;
import com.flute.haflute.common.ClusterContext;

/**
 * 集群状态同步服务, BDM和SM同DM之间同步
 */
public class ClusterStatusSynchronizer {
	private static Logger logger = LoggerFactory
			.getLogger(ClusterStatusSynchronizer.class);

	/** watcher DM上的同步辅助服务，这里取得的是其远程对象，CAJO通信实现 */
	private Object watcher;

	/** MAX_TRIES 每次同步最多尝试次数 */
	private final int MAX_TRIES = 3;

	public ClusterStatusSynchronizer() {
		long delay = 1000;
		try {
			connect2Watcher();
		} catch (Exception e) {
			delay = 60 * 1000;
		}

		if (AgentContext.getNodeRole() == ROLE.BDM
				|| AgentContext.getNodeRole() == ROLE.SLAVE) {
			Synchronizer synchronizer = null;
			if (AgentContext.getNodeRole() == ROLE.BDM) {
				synchronizer = new Synchronizer4BDM();
			} else {
				synchronizer = new Synchronizer4SM();
			}

			new Timer().schedule(synchronizer, delay, 20 * 60 * 1000);
		}

	}

	/**
	 * 查找DM上的同步辅助服务
	 * 
	 * @throws Exception
	 */
	private void connect2Watcher() throws Exception {
		try {
			watcher = Remote.getItem("//" + ClusterContext.getDM() + "/"
					+ ClusterConstants.TOPO_WATCHER);
		} catch (Exception e) {
			logger.error("can not connect to cluster status watcher", e);
			throw e;
		}
	}

	/**
	 * 同步服务执行者的抽象基类
	 * 
	 */
	private abstract class Synchronizer extends TimerTask {

	}

	/**
	 * SM的同步服务执行者
	 * 
	 */
	private class Synchronizer4SM extends Synchronizer {

		/*
		 * @see java.util.TimerTask#run()
		 */
		@Override
		public void run() {
			sync();
		}

		private void sync() {
			if (watcher == null) {
				try {
					connect2Watcher();
				} catch (Exception e) {
					logger.error("can not connect to cluster status watcher", e);
					return;
				}
			}

			try {
				Boolean ready = (Boolean) Remote.invoke(watcher, "isReady",
						new Node(ClusterContext.getLocalAddress()
								.getHostAddress()));
				if (ready) {
					Map<Node, ROLE> clusterStateDB = ClusterTopoDBService
							.getInstance().getNodeRoleDB();
					Boolean success = (Boolean) Remote.invoke(watcher,
							"update", clusterStateDB);
					int tryTimes = 1;
					while (!success && tryTimes < MAX_TRIES) {
						success = (Boolean) Remote.invoke(watcher, "update",
								clusterStateDB);
						tryTimes++;
					}
					if (!success) {
						logger.error("can NOT sync my topo db to DM:{}"
								+ ClusterContext.getDM());
					}
				}
			} catch (Exception e) {
				logger.error("can not communicate to watcher", e);
			}
		}
	}

	/**
	 * BDM的同步服务执行者
	 * 
	 */
	private class Synchronizer4BDM extends Synchronizer {

		/*
		 * @see java.util.TimerTask#run()
		 */
		@Override
		public void run() {
			sync();
		}

		@SuppressWarnings("unchecked")
		private void sync() {
			if (watcher == null) {
				try {
					connect2Watcher();
				} catch (Exception e) {
					logger.error("can not connect to cluster status watcher", e);
					return;
				}
			}

			try {
				Boolean ready = (Boolean) Remote.invoke(watcher, "isReady",
						new Node(ClusterContext.getLocalAddress()
								.getHostAddress()));
				if (ready) {
					ClusterTopoDBService topoService = ClusterTopoDBService
							.getInstance();
					Map<Node, ROLE> db = (Map<Node, ROLE>) Remote.invoke(
							watcher, "getStatus", null);
					for (Entry<Node, ROLE> entry : db.entrySet()) {
						topoService.updateNodeRole(entry.getKey(),
								entry.getValue());
					}
				}
			} catch (Exception e) {
				logger.error("can not communicate to watcher", e);
			}
		}
	}

}
